const client = require('@/db/redis');

module.exports = {
    // 创建消费组
    createConsumerGroup: async (stream, group) => {
        try {
            await client.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
        } catch (e) {
            if (!e.message.includes('BUSYGROUP')) {
                throw e; // 忽略"消费组已存在"错误
            }
        }
    },

    // 消费消息核心逻辑
    consumeMessages: async (stream, group, processor) => {
        const consumerId = `consumer_${process.pid}`; // 用PID标识唯一消费者
        
        while (true) {
            try {
                // 阻塞读取消息
                const result = await client.xreadgroup(
                    'GROUP', group, consumerId,
                    'BLOCK', 3000, // 3秒阻塞
                    'COUNT', 1,    // 每次取1条
                    'STREAMS', stream, '>' // 从新消息开始
                );
                
                if (result) {
                    const [messages] = result[0][1];
                    for (const [id, fields] of messages) {
                        const message = fields[1]; // 实际消息内容
                        const shouldAck = await processor(message);
                        
                        // 根据处理结果ACK
                        if (shouldAck) {
                            await client.xack(stream, group, id);
                        }
                    }
                }
            } catch (error) {
                logger.error('消费异常', error);
                await new Promise(res => setTimeout(res, 5000)); // 错误时暂停5秒
            }
        }
    }
};